package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Arrays;
import java.util.concurrent.ExecutorService;

/*
RichAsyncFunction:Flink提供的异步方法类，此处因为是查询操作输入类和返回类一致
实现两个方法：
    open用于初始化异步线程池
    asyncInvoke 是核心方法，里面的操作必须是异步的，如果你查询的数据库有异步api
    也可以用线程的异步方法，如果没有异步方法，就要自己利用线程池等方法实现异步查询

    维度关联： 可以有很多维度关联，所以下面采取泛型方式以便都实现维度关联


 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements DimJoinFunction<T> {
    ExecutorService executorService = null;

    public String tableName =null;

    public DimAsyncFunction(String tableName){
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("获得线程池");
        executorService = ThreadPoolUtil.getInstance();
    }

    /**
     *
     * @param obj  流中输入的数据。从obj中获取输入的key，然后和维度表中的字段去关联
     * @param resultFuture 异步请求处理之后，返回结果； 需要一个线程池对象
     * @throws Exception
     */
    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
        //submit 具体执行Runnable方法
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try{
                    long start = System.currentTimeMillis();
                    //从流对象中后去主键
                    String key = getKey(obj);
                    // 根据主键获取维度对象数据
                    JSONObject dimJsonObject = DimUtil.getDimInfo(tableName,key);
                    System.out.println("dimJsonObject: "+ dimJsonObject);
                    if(dimJsonObject != null){
                        //TODO  需要实际实现的方法，从维表中获取维度数据，补充到宽表数据中   维度数据和流数据关联，
                        join(obj,dimJsonObject);
                    }
                    System.out.println("obj: "+obj);
                    long end = System.currentTimeMillis();
                    System.out.println("异步查询耗时： " +(end-start) +"毫秒");
                    resultFuture.complete(Arrays.asList(obj));
                }catch (Exception e){
                    System.out.println(String.format(tableName +"异步查询异常： %s",e));
                    e.printStackTrace();
                }
            }
        });
    }

    //超时处理
    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {

    }
}
